package com.snda.dw.lidejia.hdfsoverftp;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.List;
import org.apache.ftpserver.ftplet.FtpFile;
import org.apache.ftpserver.ftplet.User;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class HdfsFtpFile implements FtpFile {

    private static final Logger LOGGER = LoggerFactory.getLogger(HdfsFtpFile.class);
    //
    private final Path path;
    private final User user;
    private final FileSystem hdfs;
    public static final String SUPERGROUP = "supergroup";

    public HdfsFtpFile(String path, User user, FileSystem dfs) {
        this.path = new Path(path);
        this.user = user;
        this.hdfs = dfs;
    }

    @Override
    public String getAbsolutePath() {
        Path fullPath = path.makeQualified(hdfs);
        return fullPath.toUri().getPath();
    }

    @Override
    public String getName() {
        return path.getName();
    }

    @Override
    public boolean isHidden() {
        return false;
    }

    @Override
    public boolean isDirectory() {
        boolean result = false;
        try {
            if (doesExist()) {
                FileStatus staus = hdfs.getFileStatus(path);
                result = staus.isDir();
            }
        } catch (IOException e) {
            LOGGER.error(String.format("Getting path status error: %s", path), e);
        }
        return result;
    }

    private FsPermission getPermissions()
            throws IOException {
        return hdfs.getFileStatus(path).getPermission();
    }

    @Override
    public boolean isFile() {
        boolean result = false;
        try {
            if (doesExist()) {
                result = hdfs.isFile(path);
            }
        } catch (IOException e) {
            LOGGER.error(String.format("Getting path status error: %s", path), e);
        }
        return result;
    }

    @Override
    public boolean doesExist() {
        boolean existed = false;
        try {
            existed = hdfs.exists(path);
        } catch (IOException ex) {
            LOGGER.error(String.format("Checking path existence error: %s ", path), ex);
        }
        return existed;
    }

    @Override
    public boolean isReadable() {
        boolean allowed = false;
        try {
            if (doesExist()) {
                FsPermission permissions = getPermissions();
                if (this.user.getName().equals(getOwnerName())) {
                    if (permissions.toString().substring(0, 1).equals("r")) {
                        LOGGER.debug(String.format("Reading permission for user [%s]: %s", user, path));
                        allowed = true;
                    }

                } else if (permissions.toString().substring(6, 7).equals("r")) {
                    LOGGER.debug(String.format("Reading permission for other [%s]: %s", user, path));
                    allowed = true;
                } else {
                    LOGGER.debug(String.format("No reading permission for user [%s]: %s", user, path));
                }
            }
        } catch (IOException e) {
            LOGGER.error(String.format("Checking path permissions error: %s", path), e);
        }
        return allowed;
    }

    private HdfsFtpFile getParent() {
        Path parent = path.getParent();
        return new HdfsFtpFile(parent.toString(), this.user, this.hdfs);
    }

    @Override
    public boolean isWritable() {
        boolean allowed = false;
        try {
            if (doesExist()) {
                FsPermission permissions = getPermissions();
                if (this.user.getName().equals(getOwnerName())) {
                    if (permissions.toString().substring(1, 2).equals("w")) {
                        LOGGER.debug(String.format("Writing permission for user [%s]: %s", user, path));
                        allowed = true;
                    }

                } else if (permissions.toString().substring(7, 8).equals("w")) {
                    LOGGER.debug(String.format("Writing permission for other [%s]: %s", user, path));
                    allowed = true;
                } else {
                    LOGGER.debug(String.format("No writing permission for user [%s]: %s", user, path));
                    allowed = false;
                }
            } else {
                allowed = getParent().isWritable();
            }
        } catch (IOException e) {
            LOGGER.error(String.format("Checking path(%s) permissions error", path), e);
        }
        //
        return allowed;
    }

    @Override
    public boolean isRemovable() {
        return isWritable();
    }

    @Override
    public String getOwnerName() {
        String name = null;
        try {
            if (doesExist()) {
                FileStatus fs = hdfs.getFileStatus(this.path);
                name = fs.getOwner();
            }
        } catch (IOException e) {
            LOGGER.error(String.format("Geting path(%s) owner error", path), e);
        }
        return name;
    }

    @Override
    public String getGroupName() {
        String name = null;
        try {
            if (doesExist()) {
                FileStatus fs = this.hdfs.getFileStatus(this.path);
                name = fs.getGroup();
            }
        } catch (IOException e) {
            LOGGER.error(String.format("Geting path(%s) group error", path), e);
        }
        return name;
    }

    @Override
    public int getLinkCount() {
        int count = 0;
        if (doesExist()) {
            count = (isDirectory()) ? 3 : 1;
        }
        return count;
    }

    @Override
    public long getLastModified() {
        long modified = 0;
        try {
            if (doesExist()) {
                FileStatus fs = this.hdfs.getFileStatus(this.path);
                modified = fs.getModificationTime();
            }
        } catch (IOException e) {
            LOGGER.error(String.format("Getting last modified error: %s", path), e);
        }
        return modified;
    }

    @Override
    public boolean setLastModified(long l) {
        return false;
    }

    @Override
    public long getSize() {
        long size = 0;
        try {
            if (doesExist()) {
                FileStatus fs = hdfs.getFileStatus(this.path);
                size = fs.getLen();
            }
        } catch (IOException e) {
            LOGGER.error(String.format("Getting size error: %s", path), e);
        }
        return size;
    }

    @Override
    public boolean mkdir() {
        boolean success = false;
        if (!isWritable()) {
            LOGGER.debug(String.format("No writing permission: %s", path));
        } else {
            try {
                hdfs.mkdirs(path);
                hdfs.setOwner(path, user.getName(), "supergroup");
                success = true;
            } catch (IOException e) {
                LOGGER.error(String.format("Making dir error: %s", path), e);
            }
        }
        return success;
    }

    @Override
    public boolean delete() {
        boolean success = false;
        if (doesExist()) {
            try {
                hdfs.delete(path, true);
                success = true;
            } catch (IOException e) {
                LOGGER.error(String.format("Deleting path error: %s", path), e);
            }
        }
        return success;
    }

    @Override
    public boolean move(FtpFile fileObject) {
        boolean success = false;
        if (doesExist()) {
            try {
                hdfs.rename(this.path, new Path(fileObject.getAbsolutePath()));
                success = true;
            } catch (IOException e) {
                LOGGER.error(String.format("Rename path error: %s", path), e);
            }
        }
        return success;
    }

    @Override
    public List<FtpFile> listFiles() {
        List<FtpFile> files = null;
        if (doesExist()) {
            if (!isReadable()) {
                LOGGER.debug(String.format("No reading permission: %s", path));
            } else {
                try {
                    FileStatus[] fileStats = hdfs.listStatus(path);

                    files = new ArrayList<FtpFile>(fileStats.length);
                    for (int i = 0; i < fileStats.length; i++) {
                        files.add(new HdfsFtpFile(fileStats[i].getPath().toString(), user, hdfs));
                    }
                } catch (IOException e) {
                    LOGGER.error(String.format("List files error: %s", path), e);
                }
            }
        }
        return files;
    }

    @Override
    public OutputStream createOutputStream(long l)
            throws IOException {
        OutputStream stream = null;
        if (!isWritable()) {
            throw new IOException(String.format("No writing permission: %s", path));
        }
        try {
            FSDataOutputStream out = hdfs.create(path, true);
            hdfs.setOwner(path, user.getName(), "supergroup");
            stream = out;
        } catch (IOException e) {
            LOGGER.error(String.format("Creating output stream error: %s", path), e);
        }
        return stream;
    }

    @Override
    public InputStream createInputStream(long l)
            throws IOException {
        InputStream stream = null;
        if (doesExist()) {
            if (!isReadable()) {
                throw new IOException(String.format("No reading permission: %s", path));
            }
            try {
                FSDataInputStream in = hdfs.open(path);
                stream = in;
            } catch (IOException e) {
                LOGGER.error(String.format("Creating input stream for path %s error", path), e);
            }
        } else {
            throw new IOException(String.format("Path does not existed: %s", path));
        }
        return stream;
    }
}